// Copyright 2019-2020 Parity Technologies (UK) Ltd.
// This file is part of Substrate.

// Substrate is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.

// Substrate is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.

// You should have received a copy of the GNU General Public License
// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.

use crate::business_layer_handle::BusinessLayerHandle;
use crate::discovery::DiscoveryConfig;
use crate::protocol::generic_proto::NotificationsSink;
use crate::protocol::{CustomMessageOutcome, Protocol};
use crate::request_responses::{Event, IfDisconnected, RequestFailure, ResponseFailure};
use crate::{
    discovery::DiscoveryBehaviour, discovery::DiscoveryOut, peer_info, protocol::event::DhtEvent,
    request_responses,
};
use bytes::Bytes;
use futures::channel::oneshot;
use libp2p::core::{Multiaddr, PeerId, PublicKey};
use libp2p::identify::Info;
use libp2p::kad::record;
use libp2p::swarm::NetworkBehaviour;
use sc_peerset::ReputationChange;
use std::borrow::Cow;
use std::collections::HashSet;
use std::time::Duration;

/// General behaviour of the network. Combines all protocols together.
#[derive(NetworkBehaviour)]
#[behaviour(out_event = "BehaviourOut")]
pub struct Behaviour<T: 'static + BusinessLayerHandle + Send> {
    protocol: Protocol<T>,
    /// Periodically pings and identifies the nodes we are connected to, and store information in a
    /// cache.
    peer_info: peer_info::PeerInfoBehaviour,
    /// Discovers nodes of the network.
    discovery: DiscoveryBehaviour,
    /// Generic request-response protocols.
    request_responses: request_responses::RequestResponsesBehaviour,
}

/// Event generated by `Behaviour`.
#[allow(clippy::large_enum_variant)]
#[derive(Debug)]
pub enum BehaviourOut {
    /// Opened a substream with the given node with the given notifications protocol.
    ///
    /// The protocol is always one of the notification protocols that have been registered.
    NotificationStreamOpened {
        /// Node we opened the substream with.
        remote: PeerId,
        /// The concerned protocol. Each protocol uses a different substream.
        protocol: Cow<'static, str>,
        /// Object that permits sending notifications to the peer.
        notifications_sink: NotificationsSink,
        generic_data: Vec<u8>,
        notif_protocols: Vec<Cow<'static, str>>,
        rpc_protocols: Vec<Cow<'static, str>>,
    },

    /// The [`NotificationsSink`] object used to send notifications with the given peer must be
    /// replaced with a new one.
    ///
    /// This event is typically emitted when a transport-level connection is closed and we fall
    /// back to a secondary connection.
    NotificationStreamReplaced {
        /// Id of the peer we are connected to.
        remote: PeerId,
        protocol: Cow<'static, str>,
        /// Replacement for the previous [`NotificationsSink`].
        notifications_sink: NotificationsSink,
    },

    /// Closed a substream with the given node. Always matches a corresponding previous
    /// `NotificationStreamOpened` message.
    NotificationStreamClosed {
        /// Node we closed the substream with.
        remote: PeerId,
        protocol: Cow<'static, str>,
    },

    /// Messages have been received on one or more notifications protocols.
    NotificationsReceived {
        remote: PeerId,
        messages: Vec<(Cow<'static, str>, Bytes)>,
    },

    /// We have received a request from a peer and answered it.
    ///
    /// This event is generated for statistics purposes.
    InboundRequest {
        /// Peer which sent us a request.
        peer: PeerId,
        /// Protocol name of the request.
        protocol: Cow<'static, str>,
        /// If `Ok`, contains the time elapsed between when we received the request and when we
        /// sent back the response. If `Err`, the error that happened.
        result: Result<Duration, ResponseFailure>,
    },

    /// A request has succeeded or failed.
    ///
    /// This event is generated for statistics purposes.
    RequestFinished {
        /// Peer that we send a request to.
        peer: PeerId,
        /// Name of the protocol in question.
        protocol: Cow<'static, str>,
        /// Duration the request took.
        duration: Duration,
        /// Result of the request.
        result: Result<(), RequestFailure>,
    },

    RandomKademliaStarted,
    /// Events generated by a DHT as a response to get_value or put_value requests as well as the
    /// request duration.
    Dht(DhtEvent, Duration),

    Discovered(PeerId),

    BannedRequest(PeerId, Duration),
    ReputationChanges {
        peer: PeerId,
        changes: Vec<ReputationChange>,
    },
    /// We have obtained identity information from a peer, including the addresses it is listening
    /// on.
    PeerIdentify {
        /// Id of the peer that has been identified.
        peer_id: PeerId,
        /// Information about the peer.
        info: Info,
    },

    /// Ignored event generated by lower layers.
    None,
}

impl<T: BusinessLayerHandle + Send> Behaviour<T> {
    /// Builds a new `Behaviour`.
    pub fn new(
        protocol: Protocol<T>,
        user_agent: String,
        local_public_key: PublicKey,
        disco_config: DiscoveryConfig,
        request_response_protocols: Vec<request_responses::ProtocolConfig>,
    ) -> Result<Self, request_responses::RegisterError> {
        Ok(Self {
            protocol,
            // debug_info: debug_info::DebugInfoBehaviour::new(user_agent, local_public_key),
            peer_info: peer_info::PeerInfoBehaviour::new(user_agent, local_public_key),
            discovery: disco_config.finish(),
            request_responses: request_responses::RequestResponsesBehaviour::new(
                request_response_protocols.into_iter(),
            )?,
        })
    }

    /// Returns the list of nodes that we know exist in the network.
    pub fn known_peers(&mut self) -> HashSet<PeerId> {
        self.discovery.known_peers()
    }

    /// Adds a hard-coded address for the given peer, that never expires.
    pub fn add_known_address(&mut self, peer_id: PeerId, addr: Multiaddr) {
        self.discovery.add_known_address(peer_id, addr)
    }

    /// Returns the number of nodes in each Kademlia kbucket for each Kademlia instance.
    ///
    /// Identifies Kademlia instances by their [`ProtocolId`] and kbuckets by the base 2 logarithm
    /// of their lower bound.
    pub fn num_entries_per_kbucket(&mut self) -> Option<Vec<(u32, usize)>> {
        self.discovery.num_entries_per_kbucket()
    }

    /// Returns the number of records in the Kademlia record stores.
    pub fn num_kademlia_records(&mut self) -> Option<usize> {
        self.discovery.num_kademlia_records()
    }

    /// Returns the total size in bytes of all the records in the Kademlia record stores.
    pub fn kademlia_records_total_size(&mut self) -> Option<usize> {
        self.discovery.kademlia_records_total_size()
    }

    pub fn get_address(&mut self, peer_id: &PeerId) -> Vec<Multiaddr> {
        self.discovery.addresses_of_peer(peer_id)
    }

    /// Returns true if we have a channel open with this node.
    pub fn is_open(&self, peer_id: &PeerId) -> bool {
        self.protocol.is_open(peer_id)
    }

    /// Borrows `self` and returns a struct giving access to the information about a node.
    ///
    /// Returns `None` if we don't know anything about this node. Always returns `Some` for nodes
    /// we're connected to, meaning that if `None` is returned then we're not connected to that
    /// node.
    pub fn node(&self, peer_id: &PeerId) -> Option<peer_info::Node> {
        self.peer_info.node(peer_id)
    }

    /// Start querying a record from the DHT. Will later produce either a `ValueFound` or a `ValueNotFound` event.
    pub fn get_value(&mut self, key: record::Key) {
        self.discovery.get_value(key);
    }

    /// Starts putting a record into DHT. Will later produce either a `ValuePut` or a `ValuePutFailed` event.
    pub fn put_value(&mut self, key: record::Key, value: Vec<u8>) {
        self.discovery.put_value(key, value);
    }

    /// Initiates sending a request.
    pub fn send_request(
        &mut self,
        target: &PeerId,
        protocol: &str,
        request: Vec<u8>,
        pending_response: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
        connect: IfDisconnected,
    ) {
        self.request_responses
            .send_request(target, protocol, request, pending_response, connect)
    }

    /// Returns a shared reference to the user protocol.
    pub fn user_protocol(&self) -> &Protocol<T> {
        &self.protocol
    }

    /// Returns a mutable reference to the user protocol.
    pub fn user_protocol_mut(&mut self) -> &mut Protocol<T> {
        &mut self.protocol
    }

    /// Add a self-reported address of a remote peer to the k-buckets of the supported
    /// DHTs (`supported_protocols`).
    pub fn add_self_reported_address_to_dht(
        &mut self,
        peer_id: &PeerId,
        supported_protocols: &[impl AsRef<[u8]>],
        addr: Multiaddr,
    ) {
        self.discovery
            .add_self_reported_address(peer_id, supported_protocols, addr);
    }
}

impl From<CustomMessageOutcome> for BehaviourOut {
    fn from(event: CustomMessageOutcome) -> Self {
        match event {
            CustomMessageOutcome::NotificationStreamOpened {
                remote,
                protocol,
                notifications_sink,
                generic_data,
                notif_protocols,
                rpc_protocols,
            } => Self::NotificationStreamOpened {
                remote,
                protocol,
                notifications_sink,
                generic_data,
                notif_protocols,
                rpc_protocols,
            },
            CustomMessageOutcome::NotificationStreamReplaced {
                remote,
                protocol,
                notifications_sink,
            } => Self::NotificationStreamReplaced {
                remote,
                protocol,
                notifications_sink,
            },
            CustomMessageOutcome::NotificationStreamClosed { remote, protocol } => {
                Self::NotificationStreamClosed { remote, protocol }
            }
            CustomMessageOutcome::NotificationsReceived { remote, messages } => {
                Self::NotificationsReceived { remote, messages }
            }
            CustomMessageOutcome::None => Self::None,
            CustomMessageOutcome::Banned(peer_id, duration) => {
                Self::BannedRequest(peer_id, duration)
            }
        }
    }
}

impl From<Event> for BehaviourOut {
    fn from(event: Event) -> Self {
        match event {
            Event::InboundRequest {
                peer,
                protocol,
                result,
            } => Self::InboundRequest {
                peer,
                protocol,
                result,
            },
            Event::RequestFinished {
                peer,
                protocol,
                duration,
                result,
            } => Self::RequestFinished {
                peer,
                protocol,
                duration,
                result,
            },
            Event::ReputationChanges { peer, changes } => Self::ReputationChanges { peer, changes },
        }
    }
}
impl From<peer_info::PeerInfoEvent> for BehaviourOut {
    fn from(event: peer_info::PeerInfoEvent) -> Self {
        let peer_info::PeerInfoEvent::Identified { peer_id, info } = event;
        Self::PeerIdentify { peer_id, info }
    }
}

impl From<DiscoveryOut> for BehaviourOut {
    fn from(out: DiscoveryOut) -> Self {
        match out {
            DiscoveryOut::UnroutablePeer(_peer_id) => {
                // Obtaining and reporting listen addresses for unroutable peers back
                // to Kademlia is handled by the `Identify` protocol, part of the
                // `PeerInfoBehaviour`. See the `NetworkBehaviourEventProcess`
                // implementation for `PeerInfoEvent`.
                Self::None
            }
            DiscoveryOut::Discovered(peer_id) => Self::Discovered(peer_id),
            DiscoveryOut::ValueFound(results, duration) => {
                Self::Dht(DhtEvent::ValueFound(results), duration)
            }
            DiscoveryOut::ValueNotFound(key, duration) => {
                Self::Dht(DhtEvent::ValueNotFound(key), duration)
            }
            DiscoveryOut::ValuePut(key, duration) => Self::Dht(DhtEvent::ValuePut(key), duration),
            DiscoveryOut::ValuePutFailed(key, duration) => {
                Self::Dht(DhtEvent::ValuePutFailed(key), duration)
            }
            DiscoveryOut::RandomKademliaStarted => Self::RandomKademliaStarted,
        }
    }
}
